package com.pw.study.flink.timer

import java.sql.Timestamp

import com.pw.study.flink.source.{WordBean, WordSource}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * @Desc: 定时器
 * @author: future
 * @since: 2022/5/16-21:02
 */
object ProcessFunctionTimer {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(2)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val ds: DataStream[WordBean] = env.addSource(new WordSource)
    val dsWatermark: DataStream[WordBean] = ds.assignTimestampsAndWatermarks(
      new BoundedOutOfOrdernessTimestampExtractor[WordBean](Time.seconds(3)) {
        override def extractTimestamp(t: WordBean): Long = {
          t.ts
        }
      }
    )
    val outputTag = new OutputTag[WordBean]("late Data")
    val mainStream: DataStream[String] = dsWatermark.keyBy(_.word).process(
      new KeyedProcessFunction[String, WordBean, String] {
        override def processElement(value: WordBean, ctx: KeyedProcessFunction[String, WordBean, String]#Context, out: Collector[String]): Unit = {
          //key
          println("key: " + ctx.getCurrentKey + " ts: " + ctx.timestamp())
          ctx.output(outputTag, value)
          println("注册定时器，注册时间为：" + new Timestamp(ctx.timerService().currentProcessingTime()) + " 水印： " + ctx.timerService().currentWatermark())
          ctx.timerService().registerEventTimeTimer(ctx.timerService().currentProcessingTime() + 5000L)
          out.collect("注册输出。。。。。main。。。。。。。。。")
        }

        override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, WordBean, String]#OnTimerContext, out: Collector[String]): Unit = {
          println("定时器触发，出发时间=== " + timestamp)

        }
      }
    )
    mainStream.getSideOutput(outputTag).print("late:")
    mainStream.print("main:")

    env.execute("jobName...")
  }
}
